package ds_industry_2025.ds.ds01.T5



import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.{LabeledPoint, Normalizer, VectorAssembler}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.ml.linalg.{SparseVector, Vector}
import org.apache.spark.ml.linalg.DenseVector

import java.util.Properties

/*
    1、根据子任务一的结果，计算出与用户id为6708的用户所购买相同商品种类最多的前10位用户id（只考虑他俩购买过多少种相同的商品，不考
    虑相同的商品买了多少次），并根据Hive的dwd库中相关表或MySQL数据库shtd_store中相关表，获取到这10位用户已购买过的商品，并剔除
    用户6708已购买的商品，通过计算这10位用户已购买的商品（剔除用户6708已购买的商品）与用户6708已购买的商品数据集中商品的余弦相似
    度累加再求均值，输出均值前5商品id作为推荐使用，将执行结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。

结果格式如下：
------------------------推荐Top5结果如下------------------------
相似度top1(商品id：1，平均相似度：0.983456)
相似度top2(商品id：71，平均相似度：0.782672)
相似度top3(商品id：22，平均相似度：0.7635246)
相似度top4(商品id：351，平均相似度：0.7335748)
相似度top5(商品id：14，平均相似度：0.522356)
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t1")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    val order_info=spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_info",conn)

    val order_detail=spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_detail",conn)

    //  todo 拿到每个用户购买商品的数据
    val user_buy_sku = order_info.join(order_detail, order_detail("order_id") === order_info("id"))
      .select("user_id", "sku_id")
      .distinct()

    //  todo 拿到用户6708用户购买的商品
    val user_6780_sku:Array[Double] = user_buy_sku.filter(col("user_id") === 5241)
      .select("sku_id")
      .map(_(0).toString.toDouble)
      .collect()

    //  todo 拿到和用户6708所购买商品种类相同种类最多的前10位用户
    val other_10_ids:Array[Long] = user_buy_sku.filter(col("user_id") !== 5241)
      .withColumn(
        "cos",
        when(col("sku_id").cast(DoubleType).isin(user_6780_sku: _*), lit(1.0)).otherwise(lit(0.0))
      )
      .groupBy("user_id")
      .agg(sum("cos").as("some"))
      .orderBy(desc("some"), asc("user_id"))
      .select("user_id")
      .map(_(0).toString.toLong)
      .limit(10)
      .collect()

    //  todo 找出相似度最高前10位用户所购买的商品
    val other_10_sku_id = user_buy_sku.filter(col("user_id").isin(other_10_ids: _*))
      .select("sku_id")
      .map(_(0).toString.toDouble)
      .collect()

    //  todo 读取子任务一生成的表格
    val sku_vector = spark.table("tzgc.t1")
    //  todo 将除了第一列id字段以外的字段全部转化为特征向量
    //  todo 因为slice切片处理会返回一个array[String]所以这里不需要添加array
    val vectorAssembler = new VectorAssembler()
      .setInputCols(sku_vector.columns.slice(1, sku_vector.columns.length))
      .setOutputCol("features")
    //  todo 定义流水线
    val dataFrame = new Pipeline()
      .setStages(Array(vectorAssembler))
      .fit(sku_vector)
      .transform(sku_vector)

   // todo LabeledPoint表示一个带标签的数据点，包含两个字段: 是为了方便机器学习做分类和回归
   //  todo 这里有一个细节：label会默认把第一个字段转化为double,即使我里面是toInt,也会帮我转化为double,这就是
   //  todo 为什么在下面筛选数据的时候不需要转化为double,也可以直接筛选(在交叉连接之后的数据赛选)
    //  label:  标签(通常是目标变量，如分类标签或回归值)  这里就是将id字段转化为label字段了
    //  features:特征向量(通常是输入特征)
    //  下面将id作为标签，features作为特征
   val mapData:Dataset[LabeledPoint]= dataFrame.select("id", "features")
     .map(
       r => {
         LabeledPoint(r(0).toString.toInt, r(1).asInstanceOf[Vector])
       }
     )

    //  todo Normalizer()是一个将特征向量进行归一化(和标准化好像差不多)
    //  Normalizer:标准化者   etP(2.0) 表示使用 L2 范数进行归一化。
    val normalizer=new Normalizer()
      .setInputCol("features")
      .setOutputCol("norm_features")
      .setP(2.0)
    //  todo 使用normalizer训练数据 进行归一化
    val normalizer_data = normalizer.transform(mapData)
      .select("label", "norm_features")

    //  todo 注册余弦相似度的udf函数
    //  breeze.linalg.functions.cosineDistance：计算两个向量之间的余弦距离。
    //  1 - cosineDistance：将余弦距离转换为余弦相似度。
    //  breeze.linalg.DenseVector(v1.values)：将 SparseVector 的值转换为 DenseVector，以便 breeze 库可以处理。
    spark.udf.register(
      "cos",
      (v1:DenseVector, v2:DenseVector)  => {
        1 - breeze.linalg.functions.cosineDistance(
          breeze.linalg.DenseVector(v1.values),breeze.linalg.DenseVector(v2.values)
          )
      }
    )

    normalizer_data.show

    //  todo 将数据进行条件自连接并根据余弦相似度算出最高的五个商品输出
    //  crossJoin:将左边dataframe的每一行和右边dataframe的每一行连接    结果就是 n * n行
    //  toDF为连接后的数据指定列名
    val result = normalizer_data.crossJoin(normalizer_data)
      .toDF("left_label", "left_norm_vector", "right_label", "right_norm_vector")
      .filter(col("left_label") !== col("right_label"))
      .withColumn("cos", expr("cos(left_norm_vector,right_norm_vector)"))
      .orderBy(desc("cos"))
      //  left_label为6708用户购买的商品id   right_label 为前10位用户购买的，并且剔除了6708用户购买的商品id
      .filter(col("left_label").isin(user_6780_sku: _*))
      .filter(!col("right_label").isin(user_6780_sku: _*) && col("right_label").isin(other_10_sku_id: _*)) //  完成了剔除
      .groupBy("right_label")
      .agg(avg("cos").as("cos"))
      .orderBy(desc("cos"))
      .limit(5)


    println("---------------推荐Top5结果如下---------------------------")
    /*
        todo 1.zipWithIndex 是一个方法，它会为集合中的每个元素附加一个索引（从 0 开始）。
              例如，如果集合是 List("a", "b", "c")，那么 zipWithIndex 会返回 List(("a", 0), ("b", 1), ("c", 2))。
             2. case (row,index)   模式匹配
              3.row.getAs[Double]("right_label").toInt   取right_label字段的值转化为double，然后转化为int

     */
    result.collect().zipWithIndex.foreach{
      case (row,index) =>
        val right_label=row.getAs[Double]("right_label").toInt
        val cos=row.getAs[Double]("cos")
        val output=s"相似度top${index+1} (商品id: ${right_label},平均相似度: ${cos})"
        println(output)
    }

    spark.close()

  }

}
